package hust.rpc.invoker;

import hust.rpc.constants.Constants;
import hust.rpc.enums.RpcResponseType;
import hust.rpc.pojo.Message;
import hust.rpc.pojo.RpcRequest;
import hust.rpc.pojo.RpcResponse;
import hust.rpc.proxy.Proxy;
import hust.rpc.utils.SingletonUtils;
import hust.rpc.utils.ThreadUtils;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class LocalInvoker {

    private static final ThreadLocal<Channel> channelThreadLocal = new ThreadLocal<>();

    public static Channel getThreadBindChannel() {
        return channelThreadLocal.get();
    }

    public static void invoke(RpcRequest request, Channel channel, byte codec) {
        ThreadUtils.execute(() -> {
            RpcResponse response = RpcResponse.builder()
                    .id(request.getId())
                    .build();
            try {
                channelThreadLocal.set(channel);
                Proxy instance = SingletonUtils.getInstance(Proxy.class);
                Object result = instance.invoke(request.getSignature(), request.getArgs());
                response.setResponseType(RpcResponseType.SUCCESS);
                response.setResult(result);
                if (channel != null && channel.isActive()) {
                    Message message = Message.builder()
                            .codec(codec)
                            .version(Constants.VERSION)
                            .messageType(Constants.RESPONSE_TYPE)
                            .data(response)
                            .build();
                    channel.writeAndFlush(message);
                }
            } catch (Exception e) {
                response.setResponseType(RpcResponseType.ERROR);
                response.setResult(e);
                log.warn("调用本地方法失败", e);
            } finally {
                channelThreadLocal.remove();
            }
        });
    }
}
